package com.ants.boot.system.socket;

import com.alibaba.fastjson.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @author 蚂蚁会花呗
 * @date 2022/11/10 14:01
 * 向web发送系统消息
 * 接口路径 ws://ip:port/websocket/system/{topic}/{keyId}
 */
@Component
@ServerEndpoint("/websocket/system/{topic}/{keyId}")
public class SystemMsgSocket {

    /**
     * 分割符
     */
    private static final String SEPARATOR_STR = "-";

    private static Logger logger = LoggerFactory.getLogger(SystemMsgSocket.class);

    /**
     * 与某个客户端的连接会话
     */
    private Session session;

    /**
     * 所属主题
     */
    private String topic;

    /**
     * 用户id
     */
    private String keyId;

    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的SystemMsgSocket对象。
     */
    private static CopyOnWriteArraySet<SystemMsgSocket> systemMsgSockets = new CopyOnWriteArraySet<>();

    /**
     * 用来存在线连接用户信息
     */
    private static ConcurrentHashMap<String,Session> sessionPool = new ConcurrentHashMap<String,Session>();

    /**
     * 连接成功调用的方法
     * @param session
     * @param keyId
     */
    @OnOpen
    public void onOpen(Session session,@PathParam("topic") String topic, @PathParam(value="keyId")String keyId){
        this.session = session;
        this.keyId = keyId;
        this.topic = topic;
        String sessionKey = topic + SEPARATOR_STR + keyId;
        systemMsgSockets.add(this);
        sessionPool.put(sessionKey,session);
    }

    /**
     * 链接关闭调用的方法
     */
    @OnClose
    public void onClose() {
        systemMsgSockets.remove(this);
        sessionPool.remove(this.topic + SEPARATOR_STR +this.keyId);
    }

    /**
     * 收到客户端消息后调用的方法
     *
     * @param message
     * @param message
     */
    @OnMessage
    public void onMessage(String message) {
        logger.info("websocket消息】收到客户端消息:"+message);
        sendAllMessage(this.topic,"收到消息了：" + message);
    }


    /**
     * 单点消息
     * @param keyId
     * @param message
     */
    public void sendOneMessage(String topic, String keyId, String message) {
        Session session = sessionPool.get(topic + SEPARATOR_STR + keyId);
        if (session != null&&session.isOpen()) {
            try {
                logger.info("【websocket消息】 单点消息:"+message);
                session.getAsyncRemote().sendText(message);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 单点消息(多人)
     * @param keyIds
     * @param message
     */
    public void sendMoreMessage(String topic, String[] keyIds, String message) {
        for(String keyId:keyIds) {
            Session session = sessionPool.get(topic + SEPARATOR_STR + keyId);
            if (session != null&&session.isOpen()) {
                try {
                    logger.info("【websocket消息】 单点消息:"+message);
                    session.getAsyncRemote().sendText(message);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 广播消息
     * @param message
     */
    public void sendAllMessage(String topic,String message) {
        logger.info("websocket消息】广播消息:"+message);
        for(SystemMsgSocket webSocket : systemMsgSockets) {
            try {
                if(webSocket.topic.equals(topic) && webSocket.session.isOpen()) {
                    JSONObject obj = new JSONObject();
                    obj.put("data",message);
                    webSocket.session.getAsyncRemote().sendText(obj.toJSONString());
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}
